package com.cpr.demo.mycraft.mq.redis.multi;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONUtil;
import com.cpr.demo.constant.MyMqConstant;
import com.cpr.demo.mycraft.mq.redis.MessageHandler;
import com.cpr.demo.mycraft.mq.redis.RedisMessage;
import com.cpr.demo.mycraft.mq.redis.RedisMessageQueueManager;
import com.cpr.demo.service.PersonCache2ServiceImpl;
import com.cpr.demo.service.mq.MqReportService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Set;

import static com.cpr.demo.constant.RedisKeyConstant.MQ_MULTI_GROUPS;
import static com.cpr.demo.constant.RedisKeyConstant.MQ_MULTI_QUEUE;

/**
 * @author Chen Panrui
 * created at 2021/10/21 13:50
 * Q: 为什么会有这个类？
 * A: 这是一个基于 redis.list实现的 广播(pub/sub)，一个message被多个consumer消费一次
 *
 * 测试demo-订阅某个queue {@link PersonCache2ServiceImpl#afterPropertiesSet()}
 * 测试demo-发消息给某个queue {@link com.cpr.demo.controller.MqTestController#send2(String, String)}
 * 测试数据保存-日志表 {@link com.cpr.demo.model.mq.MqReport}
 */
@Slf4j
@Component("multiRedisMQ")
public class MultiRedisMessageQueueManagerImpl implements RedisMessageQueueManager {

    @Autowired
    private String serverId;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Qualifier("singleRedisMQ")
    @Autowired
    private RedisMessageQueueManager singleRedisMq;
    @Autowired
    private MqReportService mqReportService;

    @Override
    public void publish(String queue, RedisMessage message) {
        String jsonMessage = JSONUtil.toJsonStr(message);
        log.info("应用:{},发布广播消息#组名={}", serverId, MQ_MULTI_GROUPS);
        boardCastMsg(queue, jsonMessage);
        mqReportService.save(fullQueueName(queue, serverId), message, serverId, MyMqConstant.MQ_ROLE_SENDER, MyMqConstant.MQ_TYPE_REDIS_SINGLE);
    }

    @Override
    public void subscribe(String queue, MessageHandler handler) {
        /*
         1、把chen:multi:msg:{queueName}里发布的消息 同步到 chen:multi:queue:{queueName}:{serverId}所有的队列里面
         2、当前这个应用 订阅 这个队列 chen:multi:queue:{queueName}:{serverId}(这里就是之前做的单播)
         */
        singleRedisMq.subscribe(queue + ":" + serverId, handler);
    }

    /**
     * 把chen:multi:msg:{queueName}里发布的消息 同步到 chen:multi:queue:{queueName}:{serverId}所有的队列里面
     * @param queue 队列名
     * @param jsonMessage 消息
     */
    private void boardCastMsg(String queue, String jsonMessage){
        Set<String> members = stringRedisTemplate.boundSetOps(MQ_MULTI_GROUPS).members();
        if(CollectionUtil.isEmpty(members)){
            log.info(">>>组内无应用");
            return;
        }
        if(jsonMessage!=null && !"".equals(jsonMessage)){
            for (String server : members){
                // 往每个server监听的队列里面 塞 消息
                stringRedisTemplate.boundListOps(fullQueueName(queue, server)).leftPush(jsonMessage);
            }
        }
    }

    private String fullQueueName(String queue, String serverId){
        return MQ_MULTI_QUEUE + queue + ":" + serverId;
    }
}
